package com.atguigu.day02;

import com.atguigu.utils.IntSource;
import com.atguigu.utils.IntStatistic;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Example10 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new IntSource())
                .map(new MapFunction<Integer, IntStatistic>() {
                    @Override
                    public IntStatistic map(Integer input) throws Exception {
                        return new IntStatistic(
                                input,
                                input,
                                input,
                                1,
                                input
                        );
                    }
                })
                .keyBy(r -> "integer")
                .reduce(new Statistics())
                .print();

        env.execute();
    }

    public static class Statistics implements ReduceFunction<IntStatistic> {
        @Override
        public IntStatistic reduce(IntStatistic acc, IntStatistic input) throws Exception {
            return new IntStatistic(
                    Math.min(acc.min, input.min),
                    Math.max(acc.max, input.max),
                    acc.sum + input.sum,
                    acc.count + input.count,
                    (acc.sum + input.sum) / (acc.count + input.count)
            );
        }
    }
}
